package com.qiaofang.collectmanage.service.impl;

import com.qiaofang.collectmanage.dao.CollectRuleDao;
import com.qiaofang.collectmanage.dao.ElementRuleDao;
import com.qiaofang.collectmanage.dao.domain.CollectRule;
import com.qiaofang.collectmanage.dao.elasticsearch.EsClient;
import com.qiaofang.collectmanage.service.MergeRegionService;
import com.qiaofang.collectmanage.service.common.utils.HttpClientUtils;
import com.qiaofang.collectmanage.stub.bean.*;
import com.qiaofang.collectmanage.stub.constants.TableConstants;
import org.apache.commons.lang3.StringUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;


import java.io.BufferedReader;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Service("mergeRegionService")
public class MergeRegionServiceImpl implements MergeRegionService {

    @Autowired
    ElementRuleDao elementRuleDao;

    @Autowired
    CollectRuleDao collectRuleDao;

    @Override
    public void mergeRegion(MergeRegionDTO mergeRegionDTO) throws Exception {
        crawlerCityBulk(mergeRegionDTO.getSource(),mergeRegionDTO.getListRule(),mergeRegionDTO.getWebCode(),mergeRegionDTO.getCollectType());
    }

    public void crawlerCityBulk(String source,String listRule,String webCode,String collectType) throws Exception {
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        Long start = System.currentTimeMillis();
        Map<String,String> czcs = new HashMap<>(100);

        List<CollectRule> collectRulesCzS =   collectRuleDao.findAllBySourceAndCollectType(source,collectType);

        for (CollectRule collectRulesCz : collectRulesCzS) {
            czcs.put(collectRulesCz.getCity(), collectRulesCz.getWebsite().replace("{}", "1"));
        }

        CountDownLatch latch = new CountDownLatch(collectRulesCzS.size());
        List<String> cityRegionList = new ArrayList<>();
        List<AddCityRegionDTO> regionBeans = new ArrayList<>();
        for (CollectRule collectRule : collectRulesCzS) {
            List<CrawlerCityInfoDTO> cityInfoBeans = new ArrayList<>();
            String wbczKdUrl = czcs.get(collectRule.getCity());
            if (StringUtils.isNotBlank(wbczKdUrl)) {
                CrawlerCityInfoDTO wbcz = new CrawlerCityInfoDTO();
                wbcz.setCityName(collectRule.getCity());
                wbcz.setUrl(czcs.get(collectRule.getCity()));
                wbcz.setSelectRule(listRule);
                wbcz.setWebCode(webCode);
                cityInfoBeans.add(wbcz);
            }

            Thread thread = new Thread(() -> {
                long startTime = System.currentTimeMillis();
                cityRegionList.add(collectRule.getCity() + ": " + crawlerCityInfo(cityInfoBeans));
                AddCityRegionDTO addCityRegionDTO = new AddCityRegionDTO();
                addCityRegionDTO.setCity(collectRule.getCity());
                addCityRegionDTO.setRegion(crawlerCityInfo(cityInfoBeans));
                regionBeans.add(addCityRegionDTO);
                latch.countDown();
            });
            thread.start();
        }
        latch.await();


        for (String cityRegion : cityRegionList) {
            System.out.println(cityRegion.toString());
        }
        cityRegionBulkParent(regionBeans);
        System.out.println(System.currentTimeMillis()-start);
    }


    public void cityRegionBulkParent(List<AddCityRegionDTO> regionBeans) throws Exception {
        System.out.println(regionBeans.size());
        List<Object> cityList = new ArrayList<>();
        List<Object> regionList = new ArrayList<>();
        List<Object> regionAliasList = new ArrayList<>();
        for (AddCityRegionDTO regionBean : regionBeans) {
            String cityTmp = regionBean.getCity();
            IndexCityDTO city = new IndexCityDTO();
            city.setName(cityTmp);
            city.setId(MD5Hash.getMD5AsHex(Bytes.toBytes(cityTmp)));
            cityList.add(city);

            String[] regions = regionBean.getRegion().split(", ");
            for(String region : regions){

                IndexCityRegionDTO indexCityRegion = new IndexCityRegionDTO();
                indexCityRegion.setCityName(cityTmp);
                indexCityRegion.setName(region);
                indexCityRegion.setId(MD5Hash.getMD5AsHex(Bytes.toBytes(indexCityRegion.getName())));
                indexCityRegion.setParent(MD5Hash.getMD5AsHex(Bytes.toBytes(cityTmp)));
                regionList.add(indexCityRegion);

                IndexCityRegionAliasDTO indexCityRegionAlias = new IndexCityRegionAliasDTO();
                indexCityRegionAlias.setCityName(cityTmp);
                indexCityRegionAlias.setName(region);
                indexCityRegionAlias.setId(MD5Hash.getMD5AsHex(Bytes.toBytes(indexCityRegionAlias.getName())));
                indexCityRegionAlias.setParent(MD5Hash.getMD5AsHex(Bytes.toBytes(cityTmp)));
                regionAliasList.add(indexCityRegionAlias);
            }
        }
        String indexName = TableConstants.INDEX_HSWH_CITY + "_prod" ;
        EsClient.addIndexBulks(indexName, TableConstants.INDEX_TYPE_CITY, cityList);
        EsClient.addIndexBulks(indexName, TableConstants.INDEX_TYPE_REGION, regionList);
        EsClient.addIndexBulks(indexName, TableConstants.INDEX_TYPE_REGION_ALIAS, regionAliasList);
    }


    private String crawlerCityInfo(List<CrawlerCityInfoDTO> cityInfoBeans) {
        Set<String> regions = new HashSet<>();
        for (CrawlerCityInfoDTO  cityInfoBean : cityInfoBeans) {
            String isBlank = HttpClientUtils.getPageTextByProxyIp(cityInfoBean.getUrl(), cityInfoBean.getWebCode());
            // String isBlank = requestResponseInfo.getResponseText();
            int num = 1;
            while(StringUtils.isBlank(isBlank)){
                isBlank = HttpClientUtils.getPageTextByProxyIp(cityInfoBean.getUrl(), cityInfoBean.getWebCode());
                num++;
                if(num>5){
                    break;
                }
            }
            if (StringUtils.isNotBlank(isBlank)) {
                Document doc = Jsoup.parse(isBlank);
                Elements elements = doc.select(cityInfoBean.getSelectRule());
                System.out.println(cityInfoBean.getUrl());
                Set<String> sources = new HashSet<>();
                Set<String> result = new HashSet<>();
                for (Element element : elements) {
                    if (StringUtils.isNotBlank(element.text())) {
                        String newRegion = element.text().trim();
                        if (!"全部".equals(newRegion) && !"不限".equals(newRegion)) {
                            sources.add(newRegion);
                            if (newRegion.length() == 3) {
                                newRegion = newRegion.replace("区", "");
                                if (newRegion.length() == 3) {
                                    newRegion = newRegion.replace("市", "");
                                    if (newRegion.length() == 3) {
                                        newRegion = newRegion.replace("县", "");
                                        if (newRegion.length() == 3) {
                                            newRegion = newRegion.replace("镇", "");
                                        }
                                    } else {
                                        regions.add(newRegion);
                                        result.add(newRegion);
                                    }
                                } else {
                                    regions.add(newRegion);
                                    result.add(newRegion);
                                }
                                regions.add(newRegion);
                                result.add(newRegion);
                            } else {
                                regions.add(newRegion);
                                result.add(newRegion);
                            }
                        }
                    }
                }
                System.out.println("source:" + cityInfoBean.getCityName() + ":" + sources.toString().replace("[", "").replace("]", "") + "-" + cityInfoBean.getUrl());
                System.out.println("result:" + cityInfoBean.getCityName() + ":" + result.toString().replace("[", "").replace("]", "") + "-" + cityInfoBean.getUrl());
            }
        }
        return regions.toString().replace("[", "").replace("]", "");
    }
}
